package cn.com.zetatech.loader.util;

import cn.com.zetatech.loader.entity.InputFileFromSeekableInputStream;
import cn.com.zetatech.loader.entity.SeekableByteArrayInputStream;
import com.jcraft.jsch.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.generic.GenericRecord;
import org.apache.parquet.avro.AvroParquetReader;
import org.apache.parquet.hadoop.ParquetReader;
import org.apache.parquet.io.InputFile;
import org.apache.parquet.io.SeekableInputStream;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author jian.chen@zetatech.com.cn
 * @Description TODO
 * @Date 2024/8/14 13:20
 */
@Slf4j
public class FileUtil {

    public static List<String> listFilesInDirectory(String host, String user, String password, int port, String directory) throws Exception {
        List<String> fileNames = new ArrayList<>();
        log.info("连接到源服务器: {}", host);

        JSch jsch = new JSch();
        Session session = jsch.getSession(user, host, port);
        session.setPassword(password);
        Properties config = new Properties();
        config.put("StrictHostKeyChecking", "no");
        session.setConfig(config);
        session.connect();

        Channel channel = session.openChannel("sftp");
        channel.connect();
        ChannelSftp sftpChannel = (ChannelSftp) channel;

        log.info("列出目录中的文件: {}", directory);
        Vector<ChannelSftp.LsEntry> files = sftpChannel.ls(directory);
        for (ChannelSftp.LsEntry entry : files) {
            SftpATTRS attrs = entry.getAttrs();
            if (!attrs.isDir()) {
                fileNames.add(entry.getFilename());
            }
        }
        sftpChannel.disconnect();
        session.disconnect();

        return fileNames;
    }

    public static boolean isFileTooLarge(String fileName, String remoteDir,
                                         String sftpHost, int sftpPort, String sftpUser, String sftpPass,long maxFileSize) {
        boolean isTooLarge = false;
        ChannelSftp channelSftp = null;
        Session session = null;
        long startTime = System.currentTimeMillis();

        try {
            JSch jsch = new JSch();
            session = jsch.getSession(sftpUser, sftpHost, sftpPort);
            session.setPassword(sftpPass);

            Properties config = new Properties();
            config.put("StrictHostKeyChecking", "no");
            session.setConfig(config);

            session.connect();
            log.info("{} - 已连接到SFTP服务器", LocalDateTime.now());
            channelSftp = (ChannelSftp) session.openChannel("sftp");
            channelSftp.connect();
            log.info("{} - 已打开SFTP通道", LocalDateTime.now());
            // Define file path
            String remoteFilePath = remoteDir + fileName;

            try {
                channelSftp.lstat(remoteFilePath);

                long fileSize = channelSftp.lstat(remoteFilePath).getSize();
                log.info("{} - 文件大小: {}", LocalDateTime.now(), fileSize);
                isTooLarge = fileSize > maxFileSize;

            } catch (SftpException e) {
                // File does not exist
                log.info("{} - 文件不存在: {}", LocalDateTime.now(), remoteFilePath);
                isTooLarge = false;  // File doesn't exist, so it's not too large
            }

        } catch (Exception e) {
            log.error("检查文件大小时发生异常", e);
        } finally {
            if (channelSftp != null && channelSftp.isConnected()) {
                channelSftp.disconnect();
                log.info("{} - SFTP通道已断开", LocalDateTime.now());
            }
            if (session != null && session.isConnected()) {
                session.disconnect();
                log.info("{} - 会话已断开", LocalDateTime.now());
            }
            long endTime = System.currentTimeMillis();
            log.info("{} - 文件大小检查完成，总耗时: {}毫秒", LocalDateTime.now(), (endTime - startTime));
        }
        return isTooLarge;
    }

    public static String moveFileToBackup(String remoteFilePath, String backupDir, String sftpHost, int sftpPort, String sftpUser, String sftpPass) {
        long startTime = System.currentTimeMillis();
        String backupFilePath = "";
        ChannelSftp channelSftp = null;
        Session session = null;

        try {
            JSch jsch = new JSch();
            session = jsch.getSession(sftpUser, sftpHost, sftpPort);
            session.setPassword(sftpPass);

            Properties config = new Properties();
            config.put("StrictHostKeyChecking", "no");
            session.setConfig(config);
            session.connect();
            log.info("{} - Connected to SFTP server for file move", LocalDateTime.now());

            channelSftp = (ChannelSftp) session.openChannel("sftp");
            channelSftp.connect();
            log.info("{} - SFTP channel opened for file move", LocalDateTime.now());

            // 确保备份目录以斜线结尾
            if (!backupDir.endsWith("/")) {
                backupDir += "/";
            }

            // 检查备份目录是否存在，若不存在则创建
            try {
                channelSftp.stat(backupDir);
            } catch (SftpException e) {
                if (e.id == ChannelSftp.SSH_FX_NO_SUCH_FILE) {
                    log.info("{} - Backup directory does not exist. Creating directory: {}", LocalDateTime.now(), backupDir);
                    channelSftp.mkdir(backupDir);
                    log.info("{} - Backup directory created: {}", LocalDateTime.now(), backupDir);
                } else {
                    throw e;
                }
            }

            backupFilePath = backupDir + remoteFilePath.substring(remoteFilePath.lastIndexOf('/') + 1);

            // 将文件移动到备份目录
            channelSftp.rename(remoteFilePath, backupFilePath);
            log.info("{} - File moved to backup directory: {}", LocalDateTime.now(), backupFilePath);

        } catch (Exception e) {
            log.error("Exception occurred while moving the file", e);
        } finally {
            if (channelSftp != null && channelSftp.isConnected()) {
                channelSftp.disconnect();
                log.info("{} - SFTP channel disconnected", LocalDateTime.now());
            }
            if (session != null && session.isConnected()) {
                session.disconnect();
                log.info("{} - Session disconnected", LocalDateTime.now());
            }
            long endTime = System.currentTimeMillis();
            log.info("{} - File move completed, total time: {} milliseconds", LocalDateTime.now(), (endTime - startTime));
        }
        return backupFilePath;
    }

    public static long  writeDataToCSVAndUpload (List<LinkedHashMap<String, String>> keyValuePairs,
                                                 String fileName,  String remoteDir, String sftpHost, int sftpPort,String sftpUser, String sftpPass) {
        ChannelSftp channelSftp = null;
        Session session = null;
        long startTime = System.currentTimeMillis();
        long fileSize = 0;
        try {
            JSch jsch = new JSch();
            session = jsch.getSession(sftpUser, sftpHost, sftpPort);
            session.setPassword(sftpPass);

            Properties config = new Properties();
            config.put("StrictHostKeyChecking", "no");
            session.setConfig(config);

            session.connect();
            log.info("{} - 已连接到SFTP服务器", LocalDateTime.now());

            channelSftp = (ChannelSftp) session.openChannel("sftp");
            channelSftp.connect();
            log.info("{} - 已打开SFTP通道", LocalDateTime.now());

            // 确保远程目录存在
            try {
                channelSftp.cd(remoteDir);
                log.info("{} - 远程目录存在: {}", LocalDateTime.now(), remoteDir);

            } catch (SftpException e) {
                //目录不存在，请创建
                String[] dirs = remoteDir.split("/");
                StringBuilder path = new StringBuilder();
                for (String dir : dirs) {
                    if (dir.length() > 0) {
                        path.append("/").append(dir);
                        try {
                            channelSftp.cd(path.toString());
                        } catch (SftpException ex) {
                            channelSftp.mkdir(path.toString());
                            channelSftp.cd(path.toString());
                            log.info("{} - 目录已创建: {}", LocalDateTime.now(), path);
                        }
                    }
                }
            }

            boolean fileExists = true;
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            try {
                channelSftp.get(remoteDir + "/" + fileName, baos);
                log.info("{} - 文件存在: {}", LocalDateTime.now(), remoteDir + "/" + fileName);

            } catch (SftpException e) {
                fileExists = false;
                log.info("{} - 文件不存在: {}", LocalDateTime.now(), remoteDir + "/" + fileName);
            }

            // 在内存中创建 CSV 内容
            StringBuilder csvContent = new StringBuilder();


            // 动态确定标头
            if (!fileExists) {
                //文件不存在，从 映射获取 动态创建标题
                if (!keyValuePairs.isEmpty()) {
                    LinkedHashMap<String, String> firstEntry = keyValuePairs.get(0);
                    String header = String.join(",", firstEntry.keySet());
                    csvContent.append(header).append('\n');
                }
            } else {
                // 文件存在，获取现有内容
                String existingContent = new String(baos.toByteArray(), StandardCharsets.UTF_8);
                String[] lines = existingContent.split("\n");

                // 如果可用，则使用现有标头
                if (lines.length > 0 && !lines[0].isEmpty()) {
                    csvContent.append(lines[0]).append('\n');  // Reuse existing header
                } else {
                    // 如果文件为空或损坏，则使用默认标头
                    LinkedHashMap<String, String> firstEntry = keyValuePairs.get(0);
                    String header = String.join(",", firstEntry.keySet());
                    csvContent.append(header).append('\n');                }

                // 在现有文件的最后一行后附加新数据
                for (int i = 1; i < lines.length; i++) {
                    csvContent.append(lines[i]).append('\n');
                }
            }

            for (Map<String, String> field : keyValuePairs) {
                String row = field.values().stream()
                        .map(value -> value == null ? "" : value)
                        .collect(Collectors.joining(","));
                csvContent.append(row).append('\n');
            }
            InputStream csvInputStream = new ByteArrayInputStream(csvContent.toString().getBytes());

            // 定义新 CSV 文件的完整路径
            String remoteFilePath = remoteDir + "/" + fileName;
            // 将 CSV 文件上传到远程目录
            channelSftp.put(csvInputStream, remoteFilePath);

            log.info("{} - 数据已写入并上传到远程文件: {}", LocalDateTime.now(), remoteFilePath);
            // 获取上传文件的大小
            fileSize = csvContent.length();
        } catch (Exception e) {
            log.error("写入数据到CSV并上传时发生异常", e);
        } finally {
            if (channelSftp != null && channelSftp.isConnected()) {
                channelSftp.disconnect();
                log.info("{} - SFTP通道已断开", LocalDateTime.now());
            }
            if (session != null && session.isConnected()) {
                session.disconnect();
                log.info("{} - 会话已断开", LocalDateTime.now());
            }
            long endTime = System.currentTimeMillis();
            log.info("{} - 数据写入和上传完成，总耗时: {}毫秒", LocalDateTime.now(), (endTime - startTime));
        }
        return fileSize;  // 返回文件大小
    }
    public static String getCSVHeader(Class<?> clazz) {
        StringBuilder header = new StringBuilder();
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            header.append(field.getName()).append(',');
        }
        // 删除尾随逗号
        if (header.length() > 0) {
            header.setLength(header.length() - 1);
        }
        return header.toString();
    }

    public static Map<String, List<String>> listLargeFilesInDirectory(String remoteDir, String sftpHost, int sftpPort, String sftpUser, String sftpPass, long maxFileSize) {
        List<String> largeFiles = new ArrayList<>();
        List<String> smallFiles = new ArrayList<>();
        Map<String,List<String>>  filesMap = new HashMap<>();
        ChannelSftp channelSftp = null;
        Session session = null;

        try {
            JSch jsch = new JSch();
            session = jsch.getSession(sftpUser, sftpHost, sftpPort);
            session.setPassword(sftpPass);

            Properties config = new Properties();
            config.put("StrictHostKeyChecking", "no");
            session.setConfig(config);

            session.connect();
            log.info("{} - Connected to SFTP server", LocalDateTime.now());
            channelSftp = (ChannelSftp) session.openChannel("sftp");
            channelSftp.connect();
            log.info("{} - SFTP channel opened", LocalDateTime.now());
            // 检查目录是否存在，如果不存在则创建
            try {
                channelSftp.cd(remoteDir);
                log.info("{} - Directory exists: {}", LocalDateTime.now(), remoteDir);
            } catch (SftpException e) {
                log.warn("{} - Directory does not exist, creating: {}", LocalDateTime.now(), remoteDir);
                channelSftp.mkdir(remoteDir);
                channelSftp.cd(remoteDir);
                log.info("{} - Directory created: {}", LocalDateTime.now(), remoteDir);
            }
            // List files in the remote directory
            Vector<ChannelSftp.LsEntry> fileList = channelSftp.ls(remoteDir);
            for (ChannelSftp.LsEntry entry : fileList) {
                // Filter out directories and check file size
                if (!entry.getAttrs().isDir()) {
                    long fileSize = channelSftp.lstat(remoteDir + "/" + entry.getFilename()).getSize();
                    if (fileSize > maxFileSize) {
                        largeFiles.add(entry.getFilename());
                    }
                    if (fileSize < maxFileSize) {
                        smallFiles.add(entry.getFilename());
                    }
                }
            }
            filesMap.put("largeFiles",largeFiles);
            filesMap.put("smallFiles",smallFiles);
            log.info("{} - Large files in directory {}: {}", LocalDateTime.now(), remoteDir, largeFiles);

        } catch (Exception e) {
            log.error("Error listing large files in directory", e);
        } finally {
            if (channelSftp != null && channelSftp.isConnected()) {
                channelSftp.disconnect();
                log.info("{} - SFTP channel disconnected", LocalDateTime.now());
            }
            if (session != null && session.isConnected()) {
                session.disconnect();
                log.info("{} - Session disconnected", LocalDateTime.now());
            }
        }
        return filesMap;
    }


    // 从SFTP服务器下载文件到内存中
    public static byte[] downloadFileFromSFTPToMemory(String host, String user, String password, int port, String remoteFile) {
        JSch jsch = new JSch();
        Session session = null;
        ChannelSftp channelSftp = null;
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();

        try {
            session = jsch.getSession(user, host, port);
            session.setPassword(password);

            Properties config = new Properties();
            config.put("StrictHostKeyChecking", "no");
            session.setConfig(config);

            session.connect();
            channelSftp = (ChannelSftp) session.openChannel("sftp");
            channelSftp.connect();

            try (InputStream inputStream = channelSftp.get(remoteFile)) {
                byte[] buffer = new byte[1024];
                int bytesRead;
                while ((bytesRead = inputStream.read(buffer)) != -1) {
                    byteArrayOutputStream.write(buffer, 0, bytesRead);
                }
            }

           log.info("文件从SFTP服务器下载成功。");
            return byteArrayOutputStream.toByteArray();
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        } finally {
            if (channelSftp != null) {
                channelSftp.disconnect();
            }
            if (session != null) {
                session.disconnect();
            }
        }
    }

    // 从内存中解析Parquet文件
    public static List<LinkedHashMap<String, String>> parseParquetFileFromMemory(byte[] fileData) {
        LinkedHashMap<String, String> dataMap = new LinkedHashMap<>();
        List<LinkedHashMap<String, String>> tempResults = new ArrayList<>();
        try (SeekableInputStream inputStream = new SeekableByteArrayInputStream(fileData)) {
            InputFile inputFile = new InputFileFromSeekableInputStream(fileData);
            try (ParquetReader<GenericRecord> reader = AvroParquetReader.<GenericRecord>builder(inputFile).build()) {
                GenericRecord record;
                while ((record = reader.read()) != null) {
                    log.info("读取记录: " + record);
                    for (String fieldName : record.getSchema().getFields().stream().map(field -> field.name()).toArray(String[]::new)) {
                        Object value = record.get(fieldName);
                        if (value instanceof ByteBuffer) {
                            ByteBuffer byteBuffer = (ByteBuffer) value;
                            String stringValue = StandardCharsets.UTF_8.decode(byteBuffer).toString();
                            dataMap.put(fieldName, stringValue);
                        } else {
                            dataMap.put(fieldName, String.valueOf(value));
                        }
                    }
                    tempResults.add(dataMap);
                }
            }
            // 输出Map内容
//            dataMap.forEach((key, value) -> log.info(key + ": " + value));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return tempResults;
    }

    // 清理内存
    public static void clearMemory(byte[] fileData) {
        // 通过将内容设置为0来清除字节数组（可选步骤）
        for (int i = 0; i < fileData.length; i++) {
            fileData[i] = 0;
        }
        log.info("内存已清理。");
    }
}
